package com.jcl.cloud.services.crawler.brain.task;



import java.util.concurrent.ArrayBlockingQueue;
import java.util.concurrent.BlockingQueue;
import java.util.concurrent.ThreadPoolExecutor;
import java.util.concurrent.TimeUnit;
import javax.annotation.PreDestroy;

import com.jcl.cloud.services.unicrawler.common.dto.GrabResult;
import com.netflix.astyanax.util.TimeUUIDUtils;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

public class WorkerDispatcher implements Runnable {
    private static final Logger log = LoggerFactory.getLogger(WorkerDispatcher.class);
    private WorkerType name;
    private int coreSize = 0;
    private int maxSize = 10;
    private int queueSize = 1000;
    private static final long ONE_DAY_SEC = 86400L;
    private long keepAliveTime = 86400L;
    private long noTaskSleepSeconds = 5L;
    private int hungrySize = 1000;
    private WorkerDispatcher.WorkerPool pool;
    private BlockingQueue<Runnable> queue;
    private WorkerService workerService;
    private WorkerHandler workerHandler;


    public WorkerDispatcher(WorkerType name, int coreSize, int maxSize, int queueSize, long keepAliveTime, long noTaskSleepSeconds, int hungrySize) {
        this.name = name;
        this.coreSize = coreSize;
        this.maxSize = maxSize;
        this.queueSize = queueSize;
        this.keepAliveTime = keepAliveTime;
        this.noTaskSleepSeconds = noTaskSleepSeconds;
        this.hungrySize = hungrySize;
    }

    public void run() {
        log.info("[{}]启动ing...", this.name);
        this.queue = new ArrayBlockingQueue(this.queueSize);
        this.pool = new WorkerDispatcher.WorkerPool(this.coreSize, this.maxSize, this.keepAliveTime, TimeUnit.SECONDS, this.queue);

        while(true) {
            while(true) {
                while(true) {
                    try {
                        if (this.queue.size() < this.hungrySize) {
                            Object object = this.workerService.doWorker(TimeUUIDUtils.getUniqueTimeUUIDinMicros().toString());
                            if (null == object) {
                                try {
                                    TimeUnit.SECONDS.sleep(this.noTaskSleepSeconds);
                                } catch (InterruptedException e) {
                                    log.info("[{}]休眠被打断!", this.name, e);
                                }
                            } else {
                                this.pool.execute(new WorkerHandlerWrapper(object, this.workerHandler));
                                Thread.sleep(1000);
                            }
                        }
                    } catch (Exception e) {
                        e.printStackTrace();
                        log.error("[{}]任务执行失败：", this.name, e);
                    }
                }
            }
        }
    }

    @PreDestroy
    public void destroy() {
        log.warn("Worker抢任务停止通知!!");
        this.pool.shutdown();
        Thread.interrupted();
    }

    public WorkerType getName() {
        return this.name;
    }

    public int getCoreSize() {
        return this.coreSize;
    }

    public int getMaxSize() {
        return this.maxSize;
    }

    public int getQueueSize() {
        return this.queueSize;
    }

    public long getKeepAliveTime() {
        return this.keepAliveTime;
    }

    public long getNoTaskSleepSeconds() {
        return this.noTaskSleepSeconds;
    }

    public int getHungrySize() {
        return this.hungrySize;
    }

    public WorkerDispatcher.WorkerPool getPool() {
        return this.pool;
    }

    public BlockingQueue<Runnable> getQueue() {
        return this.queue;
    }

    public WorkerService getWorkerService() {
        return this.workerService;
    }

    public WorkerHandler getWorkerHandler() {
        return this.workerHandler;
    }

    public void setName(WorkerType name) {
        this.name = name;
    }

    public void setCoreSize(int coreSize) {
        this.coreSize = coreSize;
    }

    public void setMaxSize(int maxSize) {
        this.maxSize = maxSize;
    }

    public void setQueueSize(int queueSize) {
        this.queueSize = queueSize;
    }

    public void setKeepAliveTime(long keepAliveTime) {
        this.keepAliveTime = keepAliveTime;
    }

    public void setNoTaskSleepSeconds(long noTaskSleepSeconds) {
        this.noTaskSleepSeconds = noTaskSleepSeconds;
    }

    public void setHungrySize(int hungrySize) {
        this.hungrySize = hungrySize;
    }

    public void setPool(WorkerDispatcher.WorkerPool pool) {
        this.pool = pool;
    }

    public void setQueue(BlockingQueue<Runnable> queue) {
        this.queue = queue;
    }

    public void setWorkerService(WorkerService workerService) {
        this.workerService = workerService;
    }

    public void setWorkerHandler(WorkerHandler workerHandler) {
        this.workerHandler = workerHandler;
    }

    private class WorkerPool extends ThreadPoolExecutor {
        public WorkerPool(int coreSize, int maxSize, long keepAlive, TimeUnit timeUnit, BlockingQueue<Runnable> queue) {
            super(coreSize, maxSize, keepAlive, timeUnit, queue);
        }

        protected void afterExecute(Runnable runnable, Throwable throwable) {
            super.afterExecute(runnable, throwable);
        }
    }
}
